package code.example.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

public class WatcherDefault implements Watcher {
    private static Logger logger = LoggerFactory.getLogger(WatcherDefault.class);
    AtomicBoolean connectFlag = new AtomicBoolean(false);
    ConcurrentLinkedQueue<Thread> threadWaiterQueue = new ConcurrentLinkedQueue<Thread>();
    @Override
    public void process(WatchedEvent event) {
        Event.KeeperState keeperState = event.getState();
        if(keeperState == Event.KeeperState.SyncConnected){//
            if(event.getType() == Event.EventType.None){
                logger.info("连接成功:{}",event.toString());
                connectFlag.compareAndSet(false,true);
                notifyAllWaiter();
            }else if(event.getType() == Event.EventType.NodeDataChanged){
                logger.info("节点数据改变:{}",event.getPath());
            }else if(event.getType() == Event.EventType.NodeDeleted){
                logger.info("节点数据被删除:{}",event.getPath());
            }else if(event.getType() == Event.EventType.NodeCreated){
                logger.info("节点被创建:{}",event.getPath());
            }else if(event.getType() == Event.EventType.NodeChildrenChanged){
                logger.info("子节点被改变:{}",event.getPath());
            }else{
                logger.info("事件触发:{}",event.toString());
            }


        }else{

            logger.info("watcher:{}",event.toString());
        }
    }

    private void notifyAllWaiter(){
        Thread t = null;
        while((t = threadWaiterQueue.poll()) != null){
            LockSupport.unpark(t);
        }
    }

    public void await(){
        if(connectFlag.get()){
            return;
        }else{
            threadWaiterQueue.add(Thread.currentThread());
            LockSupport.park();
        }
    }

}
